package ldh.im.rpc.connection.socket;

import ldh.im.rpc.buffer.RpcBuffer;
import ldh.im.rpc.codec.RpcDataCodec;
import ldh.im.rpc.connection.RpcDataChannel;
import ldh.im.rpc.dto.RpcData;
import ldh.im.rpc.exception.RpcErrorCode;
import ldh.im.rpc.exception.RpcException;

import java.io.IOException;
import java.net.Socket;
import java.net.SocketException;

public class RpcDataSocketChannel extends RpcSocketChannel implements RpcDataChannel {

    private RpcDataCodec rpcDataCodec;

    public RpcDataSocketChannel(RpcBuffer rpcBuffer, Socket socket, RpcDataCodec rpcDataCodec) {
        super(rpcBuffer, socket);
        this.rpcDataCodec = rpcDataCodec;
    }

    @Override
    public <T> void write(RpcData<T> rpcData) throws RpcException {
        rpcBuffer.markWriteIdx();
        try {
            rpcDataCodec.encode(rpcBuffer, rpcData);
        } catch (Exception e) {
            rpcBuffer.resetMarkWriteIdx();
            throw new RpcException(RpcErrorCode.CodecException, e);
        }
        try {
            rpcBuffer.writeToStream(this.socket.getOutputStream());
        } catch (IOException ioException) {
            throw new RpcException(RpcErrorCode.SocketIoException, ioException);
        }
    }

    @Override
    public void read() throws RpcException {
        int readSize = 0;
        try {
            readSize = rpcBuffer.readFromStream(socket.getInputStream(), 4);
        } catch (IOException ioException) {
            throw new RpcException(RpcErrorCode.SocketIoException, ioException);
        }

        if (readSize < 0) {
            throw new RpcException(RpcErrorCode.SocketIoException, new SocketException("已经关闭"));
        }

        if (rpcBuffer.readable() >= 4) {
            rpcBuffer.markReadIdx();
            int dataLength = rpcBuffer.readInt();
            if (dataLength < rpcBuffer.readable()) {
                return;
            } else {
                rpcBuffer.resetMarkReadIdx();
                readBytes(rpcBuffer, dataLength);
            }
        }
    }

    private void readBytes(RpcBuffer rpcBuffer, int length) {
        int readSize = 0;
        try {
            readSize = rpcBuffer.readFromStream(socket.getInputStream(), length);
        } catch (IOException ioException) {
            throw new RpcException(RpcErrorCode.SocketIoException, ioException);
        }

        if (readSize < 0) {
            throw new RpcException(RpcErrorCode.SocketIoException, new SocketException("已经关闭"));
        }
    }
}
